/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.core;



import com.bff.gaia.unified.runners.core.GroupByKeyViaGroupByKeyOnly.GroupByKeyOnly;

import com.bff.gaia.unified.runners.core.ReduceFnContextFactory.StateStyle;

import com.bff.gaia.unified.runners.core.StateNamespaces.WindowNamespace;

import com.bff.gaia.unified.runners.core.TimerInternals.TimerData;

import com.bff.gaia.unified.runners.core.triggers.ExecutableTriggerStateMachine;

import com.bff.gaia.unified.runners.core.triggers.TriggerStateMachineContextFactory;

import com.bff.gaia.unified.runners.core.triggers.TriggerStateMachineRunner;

import com.bff.gaia.unified.sdk.metrics.Counter;

import com.bff.gaia.unified.sdk.metrics.Metrics;

import com.bff.gaia.unified.sdk.options.PipelineOptions;

import com.bff.gaia.unified.sdk.state.TimeDomain;

import com.bff.gaia.unified.sdk.transforms.windowing.BoundedWindow;

import com.bff.gaia.unified.sdk.transforms.windowing.PaneInfo;

import com.bff.gaia.unified.sdk.transforms.windowing.PaneInfo.Timing;

import com.bff.gaia.unified.sdk.transforms.windowing.TimestampCombiner;

import com.bff.gaia.unified.sdk.transforms.windowing.Window;

import com.bff.gaia.unified.sdk.transforms.windowing.Window.ClosingBehavior;

import com.bff.gaia.unified.sdk.transforms.windowing.WindowFn;

import com.bff.gaia.unified.sdk.util.WindowTracing;

import com.bff.gaia.unified.sdk.util.WindowedValue;

import com.bff.gaia.unified.sdk.values.KV;

import com.bff.gaia.unified.sdk.values.PCollection;

import com.bff.gaia.unified.sdk.values.WindowingStrategy;

import com.bff.gaia.unified.sdk.values.WindowingStrategy.AccumulationMode;

import com.bff.gaia.unified.vendor.guava.com.google.common.annotations.VisibleForTesting;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.FluentIterable;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.ImmutableSet;

import org.joda.time.Duration;

import org.joda.time.Instant;



import javax.annotation.Nullable;

import java.util.*;



import static com.bff.gaia.unified.vendor.guava.com.google.common.base.Preconditions.checkArgument;

import static com.bff.gaia.unified.vendor.guava.com.google.common.base.Preconditions.checkState;



/**

 * Manages the execution of a {@link ReduceFn} after a {@link GroupByKeyOnly} has partitioned the

 * {@link PCollection} by key.

 *

 * <p>The {@link #onTrigger} relies on a {@link TriggerStateMachineRunner} to manage the execution

 * of the triggering logic. The {@code ReduceFnRunner}s responsibilities are:

 *

 * <ul>

 *   <li>Tracking the windows that are active (have buffered data) as elements arrive and triggers

 *       are fired.

 *   <li>Holding the watermark based on the timestamps of elements in a pane and releasing it when

 *       the trigger fires.

 *   <li>Calling the appropriate callbacks on {@link ReduceFn} based on trigger execution, timer

 *       firings, etc, and providing appropriate contexts to the {@link ReduceFn} for actions such

 *       as output.

 *   <li>Scheduling garbage collection of state associated with a specific window, and making that

 *       happen when the appropriate timer fires.

 * </ul>

 *

 * @param <K> The type of key being processed.

 * @param <InputT> The type of values associated with the key.

 * @param <OutputT> The output type that will be produced for each key.

 * @param <W> The type of windows this operates on.

 */

public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> {



  /**

   * The {@link ReduceFnRunner} depends on most aspects of the {@link WindowingStrategy}.

   *

   * <ul>

   *   <li>It runs the trigger from the {@link WindowingStrategy}.

   *   <li>It merges windows according to the {@link WindowingStrategy}.

   *   <li>It chooses how to track active windows and clear out expired windows according to the

   *       {@link WindowingStrategy}, based on the allowed lateness and whether windows can merge.

   *   <li>It decides whether to emit empty final panes according to whether the {@link

   *       WindowingStrategy} requires it.

   *   <li>

   *   <li>It uses discarding or accumulation mode according to the {@link WindowingStrategy}.

   * </ul>

   */

  private final WindowingStrategy<Object, W> windowingStrategy;



  private final OutputWindowedValue<KV<K, OutputT>> outputter;



  private final StateInternals stateInternals;



  private final Counter droppedDueToClosedWindow;



  public static final String DROPPED_DUE_TO_CLOSED_WINDOW = "droppedDueToClosedWindow";



  private final K key;



  /**

   * Track which windows are still active and the 'state address' windows which hold their state.

   *

   * <ul>

   *   <li>State: Global map for all active windows for this computation and key.

   *   <li>Lifetime: Cleared when no active windows need to be tracked. A window lives within the

   *       active window set until its trigger is closed or the window is garbage collected.

   * </ul>

   */

  private final ActiveWindowSet<W> activeWindows;



  /**

   * Always a {@link SystemReduceFn}.

   *

   * <ul>

   *   <li>State: A bag of accumulated values, or the intermediate result of a combiner.

   *   <li>State style: RENAMED

   *   <li>Merging: Concatenate or otherwise combine the state from each merged window.

   *   <li>Lifetime: Cleared when a pane fires if DISCARDING_FIRED_PANES. Otherwise cleared when

   *       trigger is finished or when the window is garbage collected.

   * </ul>

   */

  private final ReduceFn<K, InputT, OutputT, W> reduceFn;



  /**

   * Manage the setting and firing of timer events.

   *

   * <ul>

   *   <li>Merging: End-of-window and garbage collection timers are cancelled when windows are

   *       merged away. Timers created by triggers are never garbage collected and are left to fire

   *       and be ignored.

   *   <li>Lifetime: Timers automatically disappear after they fire.

   * </ul>

   */

  private final TimerInternals timerInternals;



  /**

   * Manage the execution and state for triggers.

   *

   * <ul>

   *   <li>State: Tracks which sub-triggers have finished, and any additional state needed to

   *       determine when the trigger should fire.

   *   <li>State style: DIRECT

   *   <li>Merging: Finished bits are explicitly managed. Other state is eagerly merged as needed.

   *   <li>Lifetime: Most trigger state is cleared when the final pane is emitted. However the

   *       finished bits are left behind and must be cleared when the window is garbage collected.

   * </ul>

   */

  private final TriggerStateMachineRunner<W> triggerRunner;



  /**

   * Store the output watermark holds for each window.

   *

   * <ul>

   *   <li>State: Bag of hold timestamps.

   *   <li>State style: RENAMED

   *   <li>Merging: Depending on {@link TimestampCombiner}, may need to be recalculated on merging.

   *       When a pane fires it may be necessary to add (back) an end-of-window or garbage

   *       collection hold.

   *   <li>Lifetime: Cleared when a pane fires or when the window is garbage collected.

   * </ul>

   */

  private final WatermarkHold<W> watermarkHold;



  private final ReduceFnContextFactory<K, InputT, OutputT, W> contextFactory;



  /**

   * Store the previously emitted pane (if any) for each window.

   *

   * <ul>

   *   <li>State: The previous {@link PaneInfo} passed to the user's {@code DoFn.ProcessElement}

   *       method, if any.

   *   <li>Style style: DIRECT

   *   <li>Merging: Always keyed by actual window, so does not depend on {@link #activeWindows}.

   *       Cleared when window is merged away.

   *   <li>Lifetime: Cleared when trigger is closed or window is garbage collected.

   * </ul>

   */

  private final PaneInfoTracker paneInfoTracker;



  /**

   * Store whether we've seen any elements for a window since the last pane was emitted.

   *

   * <ul>

   *   <li>State: Unless DISCARDING_FIRED_PANES, a count of number of elements added so far.

   *   <li>State style: RENAMED.

   *   <li>Merging: Counts are summed when windows are merged.

   *   <li>Lifetime: Cleared when pane fires or window is garbage collected.

   * </ul>

   */

  private final NonEmptyPanes<K, W> nonEmptyPanes;



  public ReduceFnRunner(

      K key,

      WindowingStrategy<?, W> windowingStrategy,

      ExecutableTriggerStateMachine triggerStateMachine,

      StateInternals stateInternals,

      TimerInternals timerInternals,

      OutputWindowedValue<KV<K, OutputT>> outputter,

      @Nullable SideInputReader sideInputReader,

      ReduceFn<K, InputT, OutputT, W> reduceFn,

      @Nullable PipelineOptions options) {

    this.key = key;

    this.timerInternals = timerInternals;

    this.paneInfoTracker = new PaneInfoTracker(timerInternals);

    this.stateInternals = stateInternals;

    this.outputter = outputter;

    this.reduceFn = reduceFn;

    this.droppedDueToClosedWindow =

        Metrics.counter(ReduceFnRunner.class, DROPPED_DUE_TO_CLOSED_WINDOW);



    @SuppressWarnings("unchecked")

    WindowingStrategy<Object, W> objectWindowingStrategy =

        (WindowingStrategy<Object, W>) windowingStrategy;

    this.windowingStrategy = objectWindowingStrategy;



    this.nonEmptyPanes = NonEmptyPanes.create(this.windowingStrategy, this.reduceFn);



    // Note this may incur I/O to load persisted window set data.

    this.activeWindows = createActiveWindowSet();



    this.contextFactory =

        new ReduceFnContextFactory<>(

            key,

            reduceFn,

            this.windowingStrategy,

            stateInternals,

            this.activeWindows,

            timerInternals,

            sideInputReader,

            options);



    this.watermarkHold = new WatermarkHold<>(timerInternals, windowingStrategy);

    this.triggerRunner =

        new TriggerStateMachineRunner<>(

            triggerStateMachine,

            new TriggerStateMachineContextFactory<>(

                windowingStrategy.getWindowFn(), stateInternals, activeWindows));

  }



  private ActiveWindowSet<W> createActiveWindowSet() {

    return windowingStrategy.getWindowFn().isNonMerging()

        ? new NonMergingActiveWindowSet<>()

        : new MergingActiveWindowSet<>(windowingStrategy.getWindowFn(), stateInternals);

  }



  @VisibleForTesting

  boolean isFinished(W window) {

    return triggerRunner.isClosed(contextFactory.base(window, StateStyle.DIRECT).state());

  }



  @VisibleForTesting

  boolean hasNoActiveWindows() {

    return activeWindows.getActiveAndNewWindows().isEmpty();

  }



  private Set<W> windowsThatAreOpen(Collection<W> windows) {

    Set<W> result = new HashSet<>();

    for (W window : windows) {

      ReduceFn<K, InputT, OutputT, W>.Context directContext =

          contextFactory.base(window, StateStyle.DIRECT);

      if (!triggerRunner.isClosed(directContext.state())) {

        result.add(window);

      }

    }

    return result;

  }



  private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception {

    Collection<W> result = new ArrayList<>();

    // Filter out timers that didn't trigger.

    for (W window : windows) {

      ReduceFn<K, InputT, OutputT, W>.Context directContext =

          contextFactory.base(window, StateStyle.DIRECT);

      if (triggerRunner.shouldFire(

          directContext.window(), directContext.timers(), directContext.state())) {

        result.add(window);

      }

    }

    return result;

  }



  /**

   * Incorporate {@code values} into the underlying reduce function, and manage holds, timers,

   * triggers, and window merging.

   *

   * <p>The general strategy is:

   *

   * <ol>

   *   <li>Use {@link WindowedValue#getWindows} (itself determined using {@link

   *       WindowFn#assignWindows}) to determine which windows each element belongs to. Some of

   *       those windows will already have state associated with them. The rest are considered NEW.

   *   <li>Use {@link WindowFn#mergeWindows} to attempt to merge currently ACTIVE and NEW windows.

   *       Each NEW window will become either ACTIVE or be discardedL. (See {@link ActiveWindowSet}

   *       for definitions of these terms.)

   *   <li>If at all possible, eagerly substitute NEW windows with their ACTIVE state address

   *       windows before any state is associated with the NEW window. In the common case that

   *       windows for new elements are merged into existing ACTIVE windows then no additional

   *       storage or merging overhead will be incurred.

   *   <li>Otherwise, keep track of the state address windows for ACTIVE windows so that their

   *       states can be merged on-demand when a pane fires.

   *   <li>Process the element for each of the windows it's windows have been merged into according

   *       to {@link ActiveWindowSet}. Processing may require running triggers, setting timers,

   *       setting holds, and invoking {@link ReduceFn#onTrigger}.

   * </ol>

   */

  public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception {

    if (!values.iterator().hasNext()) {

      return;

    }



    // Determine all the windows for elements.

    Set<W> windows = collectWindows(values);

    // If an incoming element introduces a new window, attempt to merge it into an existing

    // window eagerly.

    Map<W, W> windowToMergeResult = mergeWindows(windows);

    if (!windowToMergeResult.isEmpty()) {

      // Update windows by removing all windows that were merged away and adding

      // the windows they were merged to. We add after completing all the

      // removals to avoid removing a window that was also added.

      List<W> addedWindows = new ArrayList<>(windowToMergeResult.size());

      for (Map.Entry<W, W> entry : windowToMergeResult.entrySet()) {

        windows.remove(entry.getKey());

        addedWindows.add(entry.getValue());

      }

      windows.addAll(addedWindows);

    }



    prefetchWindowsForValues(windows);



    // All windows that are open before element processing may need to fire.

    Set<W> windowsToConsider = windowsThatAreOpen(windows);



    // Process each element, using the updated activeWindows determined by mergeWindows.

    for (WindowedValue<InputT> value : values) {

      processElement(windowToMergeResult, value);

    }



    // Now that we've processed the elements, see if any of the windows need to fire.

    // Prefetch state necessary to determine if the triggers should fire.

    for (W mergedWindow : windowsToConsider) {

      triggerRunner.prefetchShouldFire(

          mergedWindow, contextFactory.base(mergedWindow, StateStyle.DIRECT).state());

    }

    // Filter to windows that are firing.

    Collection<W> windowsToFire = windowsThatShouldFire(windowsToConsider);

    // Prefetch windows that are firing.

    for (W window : windowsToFire) {

      prefetchEmit(

          contextFactory.base(window, StateStyle.DIRECT),

          contextFactory.base(window, StateStyle.RENAMED));

    }

    // Trigger output from firing windows.

    for (W window : windowsToFire) {

      emit(

          contextFactory.base(window, StateStyle.DIRECT),

          contextFactory.base(window, StateStyle.RENAMED));

    }



    // We're all done with merging and emitting elements so can compress the activeWindow state.

    // Any windows which are still NEW must have come in on a new element which was then discarded

    // due to the window's trigger being closed. We can thus delete them.

    activeWindows.cleanupTemporaryWindows();

  }



  public void persist() {

    activeWindows.persist();

  }



  /** Extract the windows associated with the values. */

  private Set<W> collectWindows(Iterable<WindowedValue<InputT>> values) throws Exception {

    Set<W> windows = new HashSet<>();

    for (WindowedValue<?> value : values) {

      for (BoundedWindow untypedWindow : value.getWindows()) {

        @SuppressWarnings("unchecked")

        W window = (W) untypedWindow;

        windows.add(window);

      }

    }

    return windows;

  }



  /**

   * Invoke merge for the given windows and return a map from windows to the merge result window.

   * Windows that were not merged are not present in the map.

   */

  private Map<W, W> mergeWindows(Set<W> windows) throws Exception {

    if (windowingStrategy.getWindowFn().isNonMerging()) {

      // Return an empty map, indicating that every window is not merged.

      return Collections.emptyMap();

    }



    Map<W, W> windowToMergeResult = new HashMap<>();

    // Collect the windows from all elements (except those which are too late) and

    // make sure they are already in the active window set or are added as NEW windows.

    for (W window : windows) {

      // For backwards compat with pre 1.4 only.

      // We may still have ACTIVE windows with multiple state addresses, representing

      // a window who's state has not yet been eagerly merged.

      // We'll go ahead and merge that state now so that we don't have to worry about

      // this legacy case anywhere else.

      if (activeWindows.isActive(window)) {

        Set<W> stateAddressWindows = activeWindows.readStateAddresses(window);

        if (stateAddressWindows.size() > 1) {

          // This is a legacy window who's state has not been eagerly merged.

          // Do that now.

          ReduceFn<K, InputT, OutputT, W>.OnMergeContext premergeContext =

              contextFactory.forPremerge(window);

          reduceFn.onMerge(premergeContext);

          watermarkHold.onMerge(premergeContext);

          activeWindows.merged(window);

        }

      }



      // Add this window as NEW if it is not currently ACTIVE.

      // If we had already seen this window and closed its trigger, then the

      // window will not be currently ACTIVE. It will then be added as NEW here,

      // and fall into the merging logic as usual.

      activeWindows.ensureWindowExists(window);

    }



    // Merge all of the active windows and retain a mapping from source windows to result windows.

    activeWindows.merge(new OnMergeCallback(windowToMergeResult));

    return windowToMergeResult;

  }



  private class OnMergeCallback implements ActiveWindowSet.MergeCallback<W> {

    private final Map<W, W> windowToMergeResult;



    OnMergeCallback(Map<W, W> windowToMergeResult) {

      this.windowToMergeResult = windowToMergeResult;

    }



    /**

     * Return the subset of {@code windows} which are currently ACTIVE. We only need to worry about

     * merging state from ACTIVE windows. NEW windows by definition have no existing state.

     */

    private List<W> activeWindows(Iterable<W> windows) {

      List<W> active = new ArrayList<>();

      for (W window : windows) {

        if (activeWindows.isActive(window)) {

          active.add(window);

        }

      }

      return active;

    }



    /**

     * Called from the active window set to indicate {@code toBeMerged} (of which only {@code

     * activeToBeMerged} are ACTIVE and thus have state associated with them) will later be merged

     * into {@code mergeResult}.

     */

    @Override

    public void prefetchOnMerge(Collection<W> toBeMerged, W mergeResult) throws Exception {

      List<W> activeToBeMerged = activeWindows(toBeMerged);

      ReduceFn<K, InputT, OutputT, W>.OnMergeContext directMergeContext =

          contextFactory.forMerge(activeToBeMerged, mergeResult, StateStyle.DIRECT);

      ReduceFn<K, InputT, OutputT, W>.OnMergeContext renamedMergeContext =

          contextFactory.forMerge(activeToBeMerged, mergeResult, StateStyle.RENAMED);



      // Prefetch various state.

      triggerRunner.prefetchForMerge(mergeResult, activeToBeMerged, directMergeContext.state());

      reduceFn.prefetchOnMerge(renamedMergeContext.state());

      watermarkHold.prefetchOnMerge(renamedMergeContext.state());

      nonEmptyPanes.prefetchOnMerge(renamedMergeContext.state());

    }



    /**

     * Called from the active window set to indicate {@code toBeMerged} (of which only {@code

     * activeToBeMerged} are ACTIVE and thus have state associated with them) are about to be merged

     * into {@code mergeResult}.

     */

    @Override

    public void onMerge(Collection<W> toBeMerged, W mergeResult) throws Exception {

      // Remember we have merged these windows.

      for (W window : toBeMerged) {

        windowToMergeResult.put(window, mergeResult);

      }



      // At this point activeWindows has NOT incorporated the results of the merge.

      List<W> activeToBeMerged = activeWindows(toBeMerged);

      ReduceFn<K, InputT, OutputT, W>.OnMergeContext directMergeContext =

          contextFactory.forMerge(activeToBeMerged, mergeResult, StateStyle.DIRECT);

      ReduceFn<K, InputT, OutputT, W>.OnMergeContext renamedMergeContext =

          contextFactory.forMerge(activeToBeMerged, mergeResult, StateStyle.RENAMED);



      // Run the reduceFn to perform any needed merging.

      reduceFn.onMerge(renamedMergeContext);



      // Merge the watermark holds.

      watermarkHold.onMerge(renamedMergeContext);



      // Merge non-empty pane state.

      nonEmptyPanes.onMerge(renamedMergeContext.state());



      // Have the trigger merge state as needed.

      triggerRunner.onMerge(

          directMergeContext.window(), directMergeContext.timers(), directMergeContext.state());



      for (W active : activeToBeMerged) {

        if (active.equals(mergeResult)) {

          // Not merged away.

          continue;

        }

        // Cleanup flavor A: Currently ACTIVE window is about to be merged away.

        // Clear any state not already cleared by the onMerge calls above.

        WindowTracing.debug("ReduceFnRunner.onMerge: Merging {} into {}", active, mergeResult);

        ReduceFn<K, InputT, OutputT, W>.Context directClearContext =

            contextFactory.base(active, StateStyle.DIRECT);

        // No need for the end-of-window or garbage collection timers.

        // We will establish a new end-of-window or garbage collection timer for the mergeResult

        // window in processElement below. There must be at least one element for the mergeResult

        // window since a new element with a new window must have triggered this onMerge.

        cancelEndOfWindowAndGarbageCollectionTimers(directClearContext);

        // We no longer care about any previous panes of merged away windows. The

        // merge result window gets to start fresh if it is new.

        paneInfoTracker.clear(directClearContext.state());

      }

    }

  }



  /**

   * Redirect element windows to the ACTIVE windows they have been merged into. The compressed

   * representation (value, {window1, window2, ...}) actually represents distinct elements (value,

   * window1), (value, window2), ... so if window1 and window2 merge, the resulting window will

   * contain both copies of the value.

   */

  private ImmutableSet<W> toMergedWindows(

      final Map<W, W> windowToMergeResult, final Collection<? extends BoundedWindow> windows) {

    return ImmutableSet.copyOf(

        FluentIterable.from(windows)

            .transform(

                untypedWindow -> {

                  @SuppressWarnings("unchecked")

                  W window = (W) untypedWindow;

                  W mergedWindow = windowToMergeResult.get(window);

                  // If the element is not present in the map, the window is unmerged.

                  return (mergedWindow == null) ? window : mergedWindow;

                }));

  }



  private void prefetchWindowsForValues(Collection<W> windows) {

    // Prefetch in each of the windows if we're going to need to process triggers

    for (W window : windows) {

      ReduceFn<K, InputT, OutputT, W>.Context directContext =

          contextFactory.base(window, StateStyle.DIRECT);

      triggerRunner.prefetchForValue(window, directContext.state());

    }

  }



  /**

   * Process an element.

   *

   * @param windowToMergeResult map of windows to merged windows. If a window is not present it is

   *     unmerged.

   * @param value the value being processed

   */

  private void processElement(Map<W, W> windowToMergeResult, WindowedValue<InputT> value)

      throws Exception {

    ImmutableSet<W> windows = toMergedWindows(windowToMergeResult, value.getWindows());



    // Process the element for each (mergeResultWindow, not closed) window it belongs to.

    for (W window : windows) {

      ReduceFn<K, InputT, OutputT, W>.ProcessValueContext directContext =

          contextFactory.forValue(

              window, value.getValue(), value.getTimestamp(), StateStyle.DIRECT);

      if (triggerRunner.isClosed(directContext.state())) {

        // This window has already been closed.

        droppedDueToClosedWindow.inc();

        WindowTracing.debug(

            "ReduceFnRunner.processElement: Dropping element at {} for key:{}; window:{} "

                + "since window is no longer active at inputWatermark:{}; outputWatermark:{}",

            value.getTimestamp(),

            key,

            window,

            timerInternals.currentInputWatermarkTime(),

            timerInternals.currentOutputWatermarkTime());

        continue;

      }



      activeWindows.ensureWindowIsActive(window);

      ReduceFn<K, InputT, OutputT, W>.ProcessValueContext renamedContext =

          contextFactory.forValue(

              window, value.getValue(), value.getTimestamp(), StateStyle.RENAMED);



      nonEmptyPanes.recordContent(renamedContext.state());

      scheduleGarbageCollectionTimer(directContext);



      // Hold back progress of the output watermark until we have processed the pane this

      // element will be included within. If the element is later than the output watermark, the

      // hold will be at GC time.

      watermarkHold.addHolds(renamedContext);



      // Execute the reduceFn, which will buffer the value as appropriate

      reduceFn.processValue(renamedContext);



      // Run the trigger to update its state

      triggerRunner.processValue(

          directContext.window(),

          directContext.timestamp(),

          directContext.timers(),

          directContext.state());



      // At this point, if triggerRunner.shouldFire before the processValue then

      // triggerRunner.shouldFire after the processValue. In other words adding values

      // cannot take a trigger state from firing to non-firing.

      // (We don't actually assert this since it is too slow.)

    }

  }



  /** A descriptor of the activation for a window based on a timer. */

  private class WindowActivation {

    public final ReduceFn<K, InputT, OutputT, W>.Context directContext;

    public final ReduceFn<K, InputT, OutputT, W>.Context renamedContext;

    // If this is an end-of-window timer then we may need to set a garbage collection timer

    // if allowed lateness is non-zero.

    public final boolean isEndOfWindow;

    // If this is a garbage collection timer then we should trigger and

    // garbage collect the window.  We'll consider any timer at or after the

    // end-of-window time to be a signal to garbage collect.

    public final boolean isGarbageCollection;



    WindowActivation(

        ReduceFn<K, InputT, OutputT, W>.Context directContext,

        ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {

      this.directContext = directContext;

      this.renamedContext = renamedContext;

      W window = directContext.window();



      // The output watermark is before the end of the window if it is either unknown

      // or it is known to be before it. If it is unknown, that means that there hasn't been

      // enough data to advance it.

      boolean outputWatermarkBeforeEOW =

          timerInternals.currentOutputWatermarkTime() == null

              || !timerInternals.currentOutputWatermarkTime().isAfter(window.maxTimestamp());



      // The "end of the window" is reached when the local input watermark (for this key) surpasses

      // it but the local output watermark (also for this key) has not. After data is emitted and

      // the output watermark hold is released, the output watermark on this key will immediately

      // exceed the end of the window (otherwise we could see multiple ON_TIME outputs)

      this.isEndOfWindow =

          timerInternals.currentInputWatermarkTime().isAfter(window.maxTimestamp())

              && outputWatermarkBeforeEOW;



      // The "GC time" is reached when the input watermark surpasses the end of the window

      // plus allowed lateness. After this, the window is expired and expunged.

      this.isGarbageCollection =

          timerInternals

              .currentInputWatermarkTime()

              .isAfter(LateDataUtils.garbageCollectionTime(window, windowingStrategy));

    }



    // Has this window had its trigger finish?

    // - The trigger may implement isClosed as constant false.

    // - If the window function does not support windowing then all windows will be considered

    // active.

    // So we must take conjunction of activeWindows and triggerRunner state.

    public boolean windowIsActiveAndOpen() {

      return activeWindows.isActive(directContext.window())

          && !triggerRunner.isClosed(directContext.state());

    }

  }



  public void onTimers(Iterable<TimerData> timers) throws Exception {

    if (!timers.iterator().hasNext()) {

      return;

    }



    // Create a reusable context for each window and begin prefetching necessary

    // state.

    Map<BoundedWindow, WindowActivation> windowActivations = new HashMap();



    for (TimerData timer : timers) {

      checkArgument(

          timer.getNamespace() instanceof WindowNamespace,

          "Expected timer to be in WindowNamespace, but was in %s",

          timer.getNamespace());

      @SuppressWarnings("unchecked")

      WindowNamespace<W> windowNamespace = (WindowNamespace<W>) timer.getNamespace();

      W window = windowNamespace.getWindow();



      WindowTracing.debug(

          "{}: Received timer key:{}; window:{}; data:{} with "

              + "inputWatermark:{}; outputWatermark:{}",

          ReduceFnRunner.class.getSimpleName(),

          key,

          window,

          timer,

          timerInternals.currentInputWatermarkTime(),

          timerInternals.currentOutputWatermarkTime());



      // Processing time timers for an expired window are ignored, just like elements

      // that show up too late. Window GC is management by an event time timer

      if (TimeDomain.EVENT_TIME != timer.getDomain() && windowIsExpired(window)) {

        continue;

      }



      // How a window is processed is a function only of the current state, not the details

      // of the timer. This makes us robust to large leaps in processing time and watermark

      // time, where both EOW and GC timers come in together and we need to GC and emit

      // the final pane.

      if (windowActivations.containsKey(window)) {

        continue;

      }



      ReduceFn<K, InputT, OutputT, W>.Context directContext =

          contextFactory.base(window, StateStyle.DIRECT);

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext =

          contextFactory.base(window, StateStyle.RENAMED);

      WindowActivation windowActivation = new WindowActivation(directContext, renamedContext);

      windowActivations.put(window, windowActivation);



      // Perform prefetching of state to determine if the trigger should fire.

      if (windowActivation.isGarbageCollection) {

        triggerRunner.prefetchIsClosed(directContext.state());

      } else {

        triggerRunner.prefetchShouldFire(directContext.window(), directContext.state());

      }

    }



    // For those windows that are active and open, prefetch the triggering or emitting state.

    for (WindowActivation timer : windowActivations.values()) {

      if (timer.windowIsActiveAndOpen()) {

        ReduceFn<K, InputT, OutputT, W>.Context directContext = timer.directContext;

        if (timer.isGarbageCollection) {

          prefetchOnTrigger(directContext, timer.renamedContext);

        } else if (triggerRunner.shouldFire(

            directContext.window(), directContext.timers(), directContext.state())) {

          prefetchEmit(directContext, timer.renamedContext);

        }

      }

    }



    // Perform processing now that everything is prefetched.

    for (WindowActivation windowActivation : windowActivations.values()) {

      ReduceFn<K, InputT, OutputT, W>.Context directContext = windowActivation.directContext;

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext = windowActivation.renamedContext;



      if (windowActivation.isGarbageCollection) {

        WindowTracing.debug(

            "{}: Cleaning up for key:{}; window:{} with inputWatermark:{}; outputWatermark:{}",

            ReduceFnRunner.class.getSimpleName(),

            key,

            directContext.window(),

            timerInternals.currentInputWatermarkTime(),

            timerInternals.currentOutputWatermarkTime());



        boolean windowIsActiveAndOpen = windowActivation.windowIsActiveAndOpen();

        if (windowIsActiveAndOpen) {

          // We need to call onTrigger to emit the final pane if required.

          // The final pane *may* be ON_TIME if no prior ON_TIME pane has been emitted,

          // and the watermark has passed the end of the window.

          @Nullable

          Instant newHold =

              onTrigger(

                  directContext,

                  renamedContext,

                  true /* isFinished */,

                  windowActivation.isEndOfWindow);

          checkState(newHold == null, "Hold placed at %s despite isFinished being true.", newHold);

        }



        // Cleanup flavor B: Clear all the remaining state for this window since we'll never

        // see elements for it again.

        clearAllState(directContext, renamedContext, windowIsActiveAndOpen);

      } else {

        WindowTracing.debug(

            "{}.onTimers: Triggering for key:{}; window:{} at {} with "

                + "inputWatermark:{}; outputWatermark:{}",

            key,

            directContext.window(),

            timerInternals.currentInputWatermarkTime(),

            timerInternals.currentOutputWatermarkTime());

        if (windowActivation.windowIsActiveAndOpen()

            && triggerRunner.shouldFire(

                directContext.window(), directContext.timers(), directContext.state())) {

          emit(directContext, renamedContext);

        }



        if (windowActivation.isEndOfWindow) {

          // If the window strategy trigger includes a watermark trigger then at this point

          // there should be no data holds, either because we'd already cleared them on an

          // earlier onTrigger, or because we just cleared them on the above emit.

          // We could assert this but it is very expensive.



          // Since we are processing an on-time firing we should schedule the garbage collection

          // timer. (If getAllowedLateness is zero then the timer event will be considered a

          // cleanup event and handled by the above).

          // Note we must do this even if the trigger is finished so that we are sure to cleanup

          // any final trigger finished bits.

          checkState(

              windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO),

              "Unexpected zero getAllowedLateness");

          Instant cleanupTime =

              LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);

          WindowTracing.debug(

              "ReduceFnRunner.onTimer: Scheduling cleanup timer for key:{}; window:{} at {} with "

                  + "inputWatermark:{}; outputWatermark:{}",

              key,

              directContext.window(),

              cleanupTime,

              timerInternals.currentInputWatermarkTime(),

              timerInternals.currentOutputWatermarkTime());

          checkState(

              !cleanupTime.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE),

              "Cleanup time %s is beyond end-of-time",

              cleanupTime);

          directContext.timers().setTimer(cleanupTime, TimeDomain.EVENT_TIME);

        }

      }

    }

  }



  /**

   * Clear all the state associated with {@code context}'s window. Should only be invoked if we know

   * all future elements for this window will be considered beyond allowed lateness. This is a

   * superset of the clearing done by {@link #emit} below since:

   *

   * <ol>

   *   <li>We can clear the trigger finished bits since we'll never need to ask if the trigger is

   *       closed again.

   *   <li>We can clear any remaining garbage collection hold.

   * </ol>

   */

  private void clearAllState(

      ReduceFn<K, InputT, OutputT, W>.Context directContext,

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext,

      boolean windowIsActiveAndOpen)

      throws Exception {

    if (windowIsActiveAndOpen) {

      // Since both the window is in the active window set AND the trigger was not yet closed,

      // it is possible we still have state.

      reduceFn.clearState(renamedContext);

      watermarkHold.clearHolds(renamedContext);

      nonEmptyPanes.clearPane(renamedContext.state());

      // These calls work irrespective of whether the window is active or not, but

      // are unnecessary if the window is not active.

      triggerRunner.clearState(

          directContext.window(), directContext.timers(), directContext.state());

      paneInfoTracker.clear(directContext.state());

    } else {

      // If !windowIsActiveAndOpen then !activeWindows.isActive (1) or triggerRunner.isClosed (2).

      // For (1), if !activeWindows.isActive then the window must be merging and has been

      // explicitly removed by emit. But in that case the trigger must have fired

      // and been closed, so this case reduces to (2).

      // For (2), if triggerRunner.isClosed then the trigger was fired and entered the

      // closed state. In that case emit will have cleared all state in

      // reduceFn, triggerRunner (except for finished bits), paneInfoTracker and activeWindows.

      // We also know nonEmptyPanes must have been unconditionally cleared by the trigger.

      // Since the trigger fired the existing watermark holds must have been cleared, and since

      // the trigger closed no new end of window or garbage collection hold will have been

      // placed by WatermarkHold.extractAndRelease.

      // Thus all the state clearing above is unnecessary.

      //

      // But(!) for backwards compatibility we must allow a pipeline to be updated from

      // an sdk version <= 1.3. In that case it is possible we have an end-of-window or

      // garbage collection hold keyed by the current window (reached via directContext) rather

      // than the state address window (reached via renamedContext).

      // However this can only happen if:

      // - We have merging windows.

      // - We are DISCARDING_FIRED_PANES.

      // - A pane has fired.

      // - But the trigger is not (yet) closed.

      if (windowingStrategy.getMode() == AccumulationMode.DISCARDING_FIRED_PANES

          && !windowingStrategy.getWindowFn().isNonMerging()) {

        watermarkHold.clearHolds(directContext);

      }

    }



    // Don't need to track address state windows anymore.

    activeWindows.remove(directContext.window());

    // We'll never need to test for the trigger being closed again.

    triggerRunner.clearFinished(directContext.state());

  }



  /** Should the reduce function state be cleared? */

  private boolean shouldDiscardAfterFiring(boolean isFinished) {

    if (isFinished) {

      // This is the last firing for trigger.

      return true;

    }

    if (windowingStrategy.getMode() == AccumulationMode.DISCARDING_FIRED_PANES) {

      // Nothing should be accumulated between panes.

      return true;

    }

    return false;

  }



  private void prefetchEmit(

      ReduceFn<K, InputT, OutputT, W>.Context directContext,

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {

    triggerRunner.prefetchShouldFire(directContext.window(), directContext.state());

    triggerRunner.prefetchOnFire(directContext.window(), directContext.state());

    triggerRunner.prefetchIsClosed(directContext.state());

    prefetchOnTrigger(directContext, renamedContext);

  }



  /** Emit if a trigger is ready to fire or timers require it, and cleanup state. */

  private void emit(

      ReduceFn<K, InputT, OutputT, W>.Context directContext,

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext)

      throws Exception {

    checkState(

        triggerRunner.shouldFire(

            directContext.window(), directContext.timers(), directContext.state()));



    // Inform the trigger of the transition to see if it is finished

    triggerRunner.onFire(directContext.window(), directContext.timers(), directContext.state());

    boolean isFinished = triggerRunner.isClosed(directContext.state());



    // Will be able to clear all element state after triggering?

    boolean shouldDiscard = shouldDiscardAfterFiring(isFinished);



    // Run onTrigger to produce the actual pane contents.

    // As a side effect it will clear all element holds, but not necessarily any

    // end-of-window or garbage collection holds.

    onTrigger(directContext, renamedContext, isFinished, false /*isEndOfWindow*/);



    // Now that we've triggered, the pane is empty.

    nonEmptyPanes.clearPane(renamedContext.state());



    // Cleanup buffered data if appropriate

    if (shouldDiscard) {

      // Cleanup flavor C: The user does not want any buffered data to persist between panes.

      reduceFn.clearState(renamedContext);

    }



    if (isFinished) {

      // Cleanup flavor D: If trigger is closed we will ignore all new incoming elements.

      // Clear state not otherwise cleared by onTrigger and clearPane above.

      // Remember the trigger is, indeed, closed until the window is garbage collected.

      triggerRunner.clearState(

          directContext.window(), directContext.timers(), directContext.state());

      paneInfoTracker.clear(directContext.state());

      activeWindows.remove(directContext.window());

    }

  }



  /** Do we need to emit? */

  private boolean needToEmit(boolean isEmpty, boolean isFinished, PaneInfo.Timing timing) {

    if (!isEmpty) {

      // The pane has elements.

      return true;

    }

    if (timing == Timing.ON_TIME

        && windowingStrategy.getOnTimeBehavior() == Window.OnTimeBehavior.FIRE_ALWAYS) {

      // This is an empty ON_TIME pane.

      return true;

    }

    if (isFinished && windowingStrategy.getClosingBehavior() == ClosingBehavior.FIRE_ALWAYS) {

      // This is known to be the final pane, and the user has requested it even when empty.

      return true;

    }

    return false;

  }



  private void prefetchOnTrigger(

      final ReduceFn<K, InputT, OutputT, W>.Context directContext,

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {

    paneInfoTracker.prefetchPaneInfo(directContext);

    watermarkHold.prefetchExtract(renamedContext);

    nonEmptyPanes.isEmpty(renamedContext.state()).readLater();

    reduceFn.prefetchOnTrigger(directContext.state());

  }



  /**

   * Run the {@link ReduceFn#onTrigger} method and produce any necessary output.

   *

   * @return output watermark hold added, or {@literal null} if none.

   */

  @Nullable

  private Instant onTrigger(

      final ReduceFn<K, InputT, OutputT, W>.Context directContext,

      ReduceFn<K, InputT, OutputT, W>.Context renamedContext,

      final boolean isFinished,

      boolean isEndOfWindow)

      throws Exception {

    // Extract the window hold, and as a side effect clear it.

    final WatermarkHold.OldAndNewHolds pair =

        watermarkHold.extractAndRelease(renamedContext, isFinished).read();

    // TODO: This isn't accurate if the elements are late. See BEAM-2262

    final Instant outputTimestamp = pair.oldHold;

    @Nullable Instant newHold = pair.newHold;



    final boolean isEmpty = nonEmptyPanes.isEmpty(renamedContext.state()).read();

    if (isEmpty

        && windowingStrategy.getClosingBehavior() == ClosingBehavior.FIRE_IF_NON_EMPTY

        && windowingStrategy.getOnTimeBehavior() == Window.OnTimeBehavior.FIRE_IF_NON_EMPTY) {

      return newHold;

    }



    Instant inputWM = timerInternals.currentInputWatermarkTime();

    if (newHold != null) {

      // We can't be finished yet.

      checkState(!isFinished, "new hold at %s but finished %s", newHold, directContext.window());

      // The hold cannot be behind the input watermark.

      checkState(

          !newHold.isBefore(inputWM), "new hold %s is before input watermark %s", newHold, inputWM);

      if (newHold.isAfter(directContext.window().maxTimestamp())) {

        // The hold must be for garbage collection, which can't have happened yet.

        checkState(

            newHold.isEqual(

                LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy)),

            "new hold %s should be at garbage collection for window %s plus %s",

            newHold,

            directContext.window(),

            windowingStrategy.getAllowedLateness());

      } else {

        // The hold must be for the end-of-window, which can't have happened yet.

        checkState(

            newHold.isEqual(directContext.window().maxTimestamp()),

            "new hold %s should be at end of window %s",

            newHold,

            directContext.window());

        checkState(

            !isEndOfWindow,

            "new hold at %s for %s but this is the watermark trigger",

            newHold,

            directContext.window());

      }

    }



    // Calculate the pane info.

    final PaneInfo pane = paneInfoTracker.getNextPaneInfo(directContext, isFinished).read();



    // Only emit a pane if it has data or empty panes are observable.

    if (needToEmit(isEmpty, isFinished, pane.getTiming())) {

      // Run reduceFn.onTrigger method.

      final List<W> windows = Collections.singletonList(directContext.window());

      ReduceFn<K, InputT, OutputT, W>.OnTriggerContext renamedTriggerContext =

          contextFactory.forTrigger(

              directContext.window(),

              pane,

              StateStyle.RENAMED,

              toOutput -> {

                // We're going to output panes, so commit the (now used) PaneInfo.

                // This is unnecessary if the trigger isFinished since the saved

                // state will be immediately deleted.

                if (!isFinished) {

                  paneInfoTracker.storeCurrentPaneInfo(directContext, pane);

                }



                // Output the actual value.

                outputter.outputWindowedValue(KV.of(key, toOutput), outputTimestamp, windows, pane);

              });



      reduceFn.onTrigger(renamedTriggerContext);

    }



    return newHold;

  }



  /**

   * Schedule a timer to garbage collect the window.

   *

   * <p>The timer:

   *

   * <ul>

   *   <li>...must be fired strictly after the expiration of the window.

   *   <li>...should be as close to the expiration as possible, to have a timely output of remaining

   *       buffered data, and GC.

   * </ul>

   */

  private void scheduleGarbageCollectionTimer(ReduceFn<?, ?, ?, W>.Context directContext) {

    Instant inputWM = timerInternals.currentInputWatermarkTime();

    Instant gcTime = LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);

    WindowTracing.trace(

        "ReduceFnRunner.scheduleGarbageCollectionTimer: Scheduling at {} for "

            + "key:{}; window:{} where inputWatermark:{}; outputWatermark:{}",

        gcTime,

        key,

        directContext.window(),

        inputWM,

        timerInternals.currentOutputWatermarkTime());

    checkState(

        !gcTime.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE),

        "Timer %s is beyond end-of-time",

        gcTime);

    directContext.timers().setTimer(gcTime, TimeDomain.EVENT_TIME);

  }



  private void cancelEndOfWindowAndGarbageCollectionTimers(

      ReduceFn<?, ?, ?, W>.Context directContext) {

    WindowTracing.debug(

        "ReduceFnRunner.cancelEndOfWindowAndGarbageCollectionTimers: Deleting timers for "

            + "key:{}; window:{} where inputWatermark:{}; outputWatermark:{}",

        key,

        directContext.window(),

        timerInternals.currentInputWatermarkTime(),

        timerInternals.currentOutputWatermarkTime());

    Instant eow = directContext.window().maxTimestamp();

    directContext.timers().deleteTimer(eow, TimeDomain.EVENT_TIME);

    Instant gc = LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);

    if (gc.isAfter(eow)) {

      directContext.timers().deleteTimer(gc, TimeDomain.EVENT_TIME);

    }

  }



  private boolean windowIsExpired(BoundedWindow w) {

    return timerInternals

        .currentInputWatermarkTime()

        .isAfter(w.maxTimestamp().plus(windowingStrategy.getAllowedLateness()));

  }

}